package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.PageViewStats;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;

//数据流： web/app -> Nginx -> 日志服务器(xx.log) -> Flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> ClickHouse
//程  序： Mock -> Flume(f1.sh) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK) -> PageViewStats10sApp -> ClickHouse(ZK)
public class PageViewStats10sApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生产环境设置为Kafka分区数

        //        //1.1 开启CheckPoint
        //        env.enableCheckpointing(5 * 60000L);
        //        env.getCheckpointConfig().setCheckpointTimeout(5 * 60000L);
        //        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000L);
        //        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
        //
        //        //1.2 指定状态后端
        //        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/xxxx/xxx"));

        //TODO 2.读取Kafka dwd_page_log 主题数据创建流
        String sourceTopic = "dwd_page_log";
        String groupId = "page_view_stats_10s_app_210927";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));

        //TODO 3.将数据转换为JavaBean对象
        SingleOutputStreamOperator<PageViewStats> pageViewStatsDS = kafkaDS.map(line -> {

            JSONObject jsonObject = JSON.parseObject(line);
            //获取公共字段
            JSONObject common = jsonObject.getJSONObject("common");
            //获取页面字段
            JSONObject page = jsonObject.getJSONObject("page");
            //获取上一跳页面信息
            String lastPageId = page.getString("last_page_id");

            long sv_ct = 0L;
            if (lastPageId == null) {
                sv_ct = 1L;
            }

            return new PageViewStats("", "",
                    page.getString("page_id"),
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L,
                    sv_ct,
                    page.getLong("during_time"),
                    jsonObject.getLong("ts"));
        });

        //TODO 4.提取事件时间生成WaterMark，分组、开窗、聚合
        SingleOutputStreamOperator<PageViewStats> pageViewStatsWithWMDS = pageViewStatsDS.assignTimestampsAndWatermarks(WatermarkStrategy.<PageViewStats>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<PageViewStats>() {
            @Override
            public long extractTimestamp(PageViewStats element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        SingleOutputStreamOperator<PageViewStats> result = pageViewStatsWithWMDS.keyBy(new KeySelector<PageViewStats, Tuple5<String, String, String, String, String>>() {
            @Override
            public Tuple5<String, String, String, String, String> getKey(PageViewStats value) throws Exception {
                return new Tuple5<>(value.getPage_id(),
                        value.getAr(),
                        value.getCh(),
                        value.getIs_new(),
                        value.getVc());
            }
        })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<PageViewStats>() {
                    @Override
                    public PageViewStats reduce(PageViewStats value1, PageViewStats value2) throws Exception {
                        value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                        value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                        value1.setDur_time(value1.getDur_time() + value2.getDur_time());
                        return value1;
                    }
                }, new WindowFunction<PageViewStats, PageViewStats, Tuple5<String, String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple5<String, String, String, String, String> key, TimeWindow window, Iterable<PageViewStats> input, Collector<PageViewStats> out) throws Exception {

                        //获取数据
                        PageViewStats pageViewStats = input.iterator().next();

                        //补充窗口信息
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        pageViewStats.setStt(sdf.format(window.getStart()));
                        pageViewStats.setEdt(sdf.format(window.getEnd()));

                        //输出数据
                        out.collect(pageViewStats);
                    }
                });

        //TODO 5.将数据写出到ClickHouse
        result.print(">>>>>>>>>>>>>");
        result.addSink(MyClickHouseUtil.getClickHouseSink("insert into dws_pv_vc_ch_isnew_ar_10s values(?,?,?,?,?,?,?,?,?,?,?)"));

        //TODO 6.启动
        env.execute("PageViewStats10sApp");

    }

}
